package bigdata.scala.spark.streaming.ycfx.usermeterstatusqueue

import java.time.LocalDateTime
import java.time.format.DateTimeFormatter
import java.util

import bigdata.java.framework.spark.kafka.{SKProcessor, SimpleSparkKafka}
import bigdata.java.framework.spark.util.client.HBaseClient
import bigdata.scala.spark.streaming.ycfx.model.UserMeterModel
import com.alibaba.fastjson.JSON
import org.apache.hadoop.hbase.TableName
import org.apache.kafka.clients.consumer.ConsumerRecord
import org.apache.spark.streaming.api.java.JavaInputDStream
import bigdata.java.framework.spark.util.TimerUtil

/**
 * @author : weixuyang
 * @date : 2020/1/1 9:15
 * @description : 用户电能表数据获取到落地
 * 参数示例 : appname: UserStatus topics:Data1.1 groupId: duration: maxnum:
 */
//定时任务：每天0时05分执行
//      TimerUtil.timerEveryDay(new UserMeterStatusQueue(),0,5）
object UserMeterStatusQueue {
  //获取Kafka数据并
  def getUserMeterStatus(args: Array[String]): Unit = {

    val simpleSparkKafka = new SimpleSparkKafka(args)
    simpleSparkKafka.execute(new SKProcessor {

      override def process(directStream: JavaInputDStream[ConsumerRecord[String, String]]): Unit = {
        val stream = directStream.inputDStream
        stream.foreachRDD{rdd=>
          if(!rdd.isEmpty()){
            val baseData = rdd.map(kafka => JSON.parseObject(kafka.value()))
              .map((json => {
                val meterId = json.getString("METER_ID")
                val collMeterId = json.getString("COLL_METER_ID")
                val wiringMode = json.getString("WIRING_MODE")
                val meterAssetno = json.getString("METER_ASSETNO")
                val collSection = json.getString("COLL_SECTION")
                val cityOrgNo = json.getString("CITY_ORG_NO")
                val cityOrgName = json.getString("CITY_ORG_NAME")
                val countyOrgNo = json.getString("COUNTY_ORG_NO")
                val countyOrgName = json.getString("COUNTY_ORG_NAME")
                val orgNo = json.getString("ORG_NO")
                val orgName = json.getString("ORG_NAME")
                val tgId = json.getString("TG_ID")
                val tgNo = json.getString("TG_NO")
                val tgName = json.getString("TG_NAME")
                val pubPrivFlag = json.getString("PUB_PRIV_FLAG")
                val consId = json.getString("CONS_ID")
                val consNo = json.getString("CONS_NO")
                val consName = json.getString("CONS_NAME")
                val lineId = json.getString("LINE_ID")
                val lineName = json.getString("LINE_NAME")
                val createTime = LocalDateTime.now().format(DateTimeFormatter.ofPattern("yyyy-MM-dd HH:mm:ss"))
                val updateTime = LocalDateTime.now().format(DateTimeFormatter.ofPattern("yyyy-MM-dd HH:mm:ss"))


                UserMeterModel(meterId,collMeterId,wiringMode,meterAssetno,collSection,cityOrgNo,cityOrgName,countyOrgNo,countyOrgName,orgNo,orgName,tgId,tgNo,tgName,pubPrivFlag,consId,consNo,consName,lineId,lineName,createTime,updateTime
                )
              }))
            //将数据落地到HBase
            baseData.foreachPartition { partitionOfRecods =>
              partitionOfRecods.foreach(t => {
                val rowkey:String =t.meterId
                val tableName:String = TableName.valueOf("").toString
                val columnFamily:String = ""
                val list = new util.ArrayList[util.Map[String,String]]()
                val map = new util.HashMap[String,String]
                map.put("meterId",t.meterId)
                map.put("collMeterId",t.collMeterId)
                map.put("wiringMode",t.wiringMode)
                map.put("meterAssetno",t.meterAssetno)
                map.put("collSection",t.collSection)
                map.put("cityOrgNo",t.cityOrgNo)
                map.put("cityOrgName",t.cityOrgName)
                map.put("countyOrgNo",t.countyOrgNo)
                map.put("cityOrgName",t.cityOrgName)
                map.put("countyOrgNo",t.countyOrgNo)
                map.put("countyOrgName",t.countyOrgName)
                map.put("orgNo",t.orgNo)
                map.put("orgName",t.orgName)
                map.put("tgId",t.tgId)
                map.put("pubPrivFlag",t.pubPrivFlag)
                map.put("tgNo",t.tgNo)
                map.put("tgName",t.tgName)
                map.put("pubPrivFlag",t.pubPrivFlag)
                map.put("consId",t.consId)
                map.put("consNo",t.consNo)
                map.put("consName",t.consName)
                map.put("createTime",t.createTime)
                map.put("lineId",t.lineId)
                map.put("lineName",t.lineName)
                map.put("createTime",t.createTime)
                map.put("updateTime",t.updateTime)
                list.add(map)
                HBaseClient.getInstance().put("tableName","meterId","columnFamily",list)

              })
            }
          }
        }
      }
    }).start()
  }


}
